package com.baske.controller;

import com.baske.bean.ProductOrder;
import com.baske.conf.JmsConfig;
import com.baske.producer.PayProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.List;
import java.util.UUID;

/**
 * RocketMQ生产消息接口
 */
@RestController
public class PayController {

    @Resource
    private PayProducer payProducer;

    // http://localhost:8080/api/v1/pay_cb?text=TestDB123
    @RequestMapping("/api/v1/pay_cb")
    public Object callback(String text) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(JmsConfig.TOPIC,"tag", UUID.randomUUID().toString(),("RocketMQ-Test:"+ text).getBytes());
        // 同步发送，需要等待响应再返回
        // SendResult sendResult = payProducer.getProducer().send(message);
        // 同步发送，指定Topic下的第一个queue发送
//        SendResult sendResult = payProducer.getProducer().send(message, new MessageQueueSelector() {
//            @Override
//            public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
//                int queueNum = Integer.parseInt(o.toString());
//                return list.get(queueNum);
//            }
//        }, 0);

        // 异步发送：指定Topic下的第二个queue发送
        payProducer.getProducer().send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                int queueNum = Integer.parseInt(o.toString());
                return list.get(queueNum);
            }
        },1, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("发送结果=%s,msg=%s",sendResult.getSendStatus(),sendResult.toString());
            }

            @Override
            public void onException(Throwable throwable) {

            }
        });
        // 单向发送：Oneway模式，无需要等待响应
        //payProducer.getProducer().sendOneway(message);
        // 异步发送：回调成功触发相对应的业务
//        payProducer.getProducer().send(message, new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//            }
//
//            @Override
//            public void onException(Throwable throwable) {
//            }
//        });

        //System.out.println(sendResult);
        return new HashMap<>();
    }


    /**
     * 顺序投递消息
     * 订单ID相同的投递到同一个Topic的同一个Queue中
     * @return
     * @throws InterruptedException
     * @throws RemotingException
     * @throws MQClientException
     * @throws MQBrokerException
     */
    // http://localhost:8080/api/v2/pay_cb
    @RequestMapping("/api/v2/pay_cb")
    public Object callback() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {

        List<ProductOrder> list = ProductOrder.getOrderList();

        for (int i = 0; i < list.size(); i++) {
            ProductOrder productOrder = list.get(i);
            Message message = new Message(JmsConfig.TOPIC_ORDER,"tag", productOrder.getOrderId()+"",productOrder.toString().getBytes(StandardCharsets.UTF_8));
            //同步发送，指定Topic下的第一个queue发送
            SendResult sendResult = payProducer.getProducer().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Integer id = (Integer) o;
                    long index = id % list.size();
                    return list.get((int)index);
                }
            }, productOrder.getOrderId());
            System.out.printf("发送结果=%s,sendResult=%s,orderId=%s,type=%s\n",sendResult.getSendStatus(),sendResult.toString(),productOrder.getOrderId(),productOrder.getType());
        }
        return new HashMap<>().put("status","消息投递成功");
    }

}
